package com.shujia.window

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo04Process {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)


    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)


    /**
     * 每隔5s统计一次单词的数量
     * 返回： 单词,数量,窗口的结束时间
     * (String,Int,Long)
     */

    val wordsKV: WindowedStream[(String, Int), String, TimeWindow] = linesDS.flatMap(_.split(","))
      .map(word => (word, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))

    wordsKV
      .process(
        /**
         * 需要四个参数：IN, OUT, KEY, W
         * IN：输入的数据类型
         * OUT：输出的数据类型
         * KEY：keyBy指定的数据类型
         * W：窗口类型
         */
        new ProcessWindowFunction[(String, Int), (String, Int, Long), String, TimeWindow] {
          /**
           * process方法 每隔5s（基于窗口的大小）会调用一次
           * Flink会将同一个key的在一个窗口（5s）内的数据 都发送给process方法进行处理
           *
           * @param key      当前处理的Key（单词）
           * @param context  上下文对象 可以获取时间、状态等环境变量
           * @param elements 当前窗口同一个Key的所有数据
           * @param out      用于将数据发送个下游
           */
          override def process(key: String, context: Context, elements: Iterable[(String, Int)], out: Collector[(String, Int, Long)]): Unit = {
            val cnt: Int = elements.size

            val endTime: Long = context.window.getEnd

            out.collect((key, cnt, endTime))
          }
        })
      .print()


    env.execute()


  }

}
